package net.bw.realtime.warehouse.DWS;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bw.realtime.warehouse.DWS
 * @Author: zhangleqing
 * @CreateTime: 2025-08-28  09:30
 * @Description: TODO
 * @Version: 1.0
 */
public class DwsCarsLog {
    public static void main(String[] args) {
        // 1.配置文件
        TableEnvironment tableEnv = getTableEnv();

        // 2.读取
        readTable(tableEnv);

        // 3.处理加工
        Table table = clean(tableEnv);

        table.execute().print();
        // 4.映射表
        // 5.保存Doris中
    }


/*



 */
    private static Table clean(TableEnvironment tableEnv) {
        Table tableResult = tableEnv.sqlQuery(
                "select * from dwd_summary_cars_log limit 10"
        );

        return tableResult;
    }

    private static void readTable(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "create table dwd_summary_cars_log(\n" +
                        "    id STRING,\n" +
                        "    owerId BIGINT,\n" +
                        "    opTime TIMESTAMP(3),\n" +
                        "    cId BIGINT,\n" +
                        "    carCode STRING,\n" +
                        "    ctype STRING,\n" +
//                        "    PRIMARY KEY (id) NOT ENFORCED,\n" +
                        "    WATERMARK FOR opTime AS opTime - INTERVAL '0' MINUTE\n" +
                        ") WITH (\n" +
                        "      'connector' = 'kafka',\n" +
                        "      'topic' = 'dwd_summary_cars_log',\n" +
                        "      'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "      'properties.group.id' = 'dwd_summary_cars_log',\n" +
                        "      'scan.startup.mode' = 'earliest-offset',\n" +
                        "      'format' = 'json'\n" +
                        ")"
        );
    }

    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        // 并行度
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }

}
